package com.quanyan.club.mq.consumer;

import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.quanyan.club.entity.db.TbClubMsgConsumerRecord;
import com.quanyan.club.entity.db.TbClubMsgConsumerRecordExample;
import com.quanyan.club.entity.db.TbClubOnlineMemberAccount;
import com.quanyan.club.entity.db.TbClubOnlineMemberAccountStream;
import com.quanyan.club.entity.db.TbClubOnlineMemberAccountStreamExample;
import com.quanyan.club.mapper.ClubAccountExtendMapper;
import com.quanyan.club.mapper.TbClubMsgConsumerRecordMapper;
import com.quanyan.club.mapper.TbClubOnlineMemberAccountStreamMapper;
import com.quanyan.club.mq.constant.MqConstant;
import com.quanyan.club.service.ClubFinanceMgmService;
import com.quanyan.common.constants.Constants;
import com.quanyan.common.utils.ObjectUtil;
import com.quanyan.mq.enums.MqConsumerStatus;
import com.quanyan.mq.mq.consumer.AbstractMqConsumer;
import com.quanyan.mq.mq.consumer.MqTopic;
import com.quanyan.orderpay.response.mq.OrderNotifyMsg;
import com.quanyan.orderpay.service.OrderPayService;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import java.math.BigDecimal;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import javax.annotation.PostConstruct;

/**
 * Created by zhang on 16/6/27.
 */
@Component
public class ClubRechargeConsumer extends AbstractMqConsumer {

    @Autowired
    private ClubFinanceMgmService clubFinanceMgmService;
    @Autowired
    private TbClubMsgConsumerRecordMapper clubMsgConsumerRecordMapper;
    @Autowired
    private ClubAccountExtendMapper clubAccountExtendMapper;
    @Autowired
    private TbClubOnlineMemberAccountStreamMapper clubOnlineMemberAccountStreamMapper;

    @Value("${mq.nameServer.address}")
    private String nameServerAddress;

    //@PostConstruct
    public void run() {
        super.start();
    }

    @Override
    @Transactional(rollbackFor = Exception.class,propagation = Propagation.REQUIRED)
    protected MqConsumerStatus process(List<MessageExt> messages, ConsumeConcurrentlyContext context) {
        MessageExt messageExt = messages.get(0);
        String msgId = messageExt.getMsgId();
        byte[] bodyByte = messageExt.getBody();
        //查看该消息是否被消费
        List<TbClubMsgConsumerRecord> clubMsgConsumerRecordList = this.getClubMsgConsumerRecord(msgId);
        if (clubMsgConsumerRecordList.isEmpty()) {
            OrderNotifyMsg orderNotifyMsg = (OrderNotifyMsg) ObjectUtil.toObject(bodyByte);
            //查看该信息是否为充值信息
            int uid = orderNotifyMsg.getUid().intValue();
            int clubId = Integer.parseInt(orderNotifyMsg.getExtraParam());
            String outTradeNo = orderNotifyMsg.getOutTradeNo();
            List<TbClubOnlineMemberAccountStream> streamList = this.clubFinanceMgmService.getClubOnlineMemberAccountStreamByOutTradeNo(
                    clubId,
                    uid,
                    outTradeNo,
                    (byte) 2);
            if (!streamList.isEmpty()) {

                /**------------------------------添加消息消费记录-------------------------------**/
                TbClubMsgConsumerRecord record = new TbClubMsgConsumerRecord();
                record.setTopic(messageExt.getTopic());
                record.setMsgId(messageExt.getMsgId());
                record.setTag(messageExt.getTags());
                record.setStatus(false);
                this.insertClubMsgComsumerRecord(record);
                /**-------------------------------消息消费记录添加结束------------------------------**/
                TbClubOnlineMemberAccountStream clubOnlineMemberAccountStream = streamList.get(0);
                //俱乐部成员账户余额增加
                TbClubOnlineMemberAccount clubOnlineMemberAccount = this.clubFinanceMgmService.getClubOnlineMemberAccount(clubId, uid);
                if (clubOnlineMemberAccount != null) {
                    BigDecimal preAmount = clubOnlineMemberAccount.getMoneyTotal();
                    BigDecimal money = orderNotifyMsg.getOrderAmount();
                    BigDecimal sufAmount = money.add(preAmount);

                    clubOnlineMemberAccountStream.setPreAmount(preAmount);
                    clubOnlineMemberAccountStream.setSufAmount(sufAmount);
                    //0:待处理 1：处理成功 2处理中 3：处理失败
                    clubOnlineMemberAccountStream.setStatus((byte)1);

                    Map map = new HashMap<>();
                    map.put("money",money);
                    map.put("uid",uid);
                    map.put("clubId",clubId);
                    int n = this.clubAccountExtendMapper.updateClubOnlineMemberAccount(map);

                    if(n>0){
                        TbClubOnlineMemberAccountStreamExample example = new TbClubOnlineMemberAccountStreamExample();
                        TbClubOnlineMemberAccountStreamExample.Criteria criteria = example.createCriteria();
                        criteria.andIdEqualTo(clubOnlineMemberAccountStream.getId());
                        criteria.andStatusEqualTo((byte)2);

                        int m = this.clubOnlineMemberAccountStreamMapper.updateByExampleSelective(clubOnlineMemberAccountStream,example);
                        if(m <= 0){

                            try {
                                throw new Exception();
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }else{
                            //更新消息消费状态
                            record.setStatus(true);
                            this.updateClubMsgComsumerRecord(record);

                            return MqConsumerStatus.CONSUMER_SUCCESS;
                        }
                    }

                }else{
                    return MqConsumerStatus.RE_CONSUME_LATER;
                }


            }


        }
        return MqConsumerStatus.RE_CONSUME_LATER;
    }

    @Override
    protected String getConsumerGroup() {
        return MqConstant.CLUB_RECHARGE_CONSUMER_GROUP_NAME;
    }

    @Override
    protected String getNameServerAddress() {
        return nameServerAddress;
    }

    @Override
    protected String getInstanceName() {
        return MqConstant.CLUB_RECHARGE_CONSUMER_INSTANCE_NAME;
    }

    @Override
    protected MqTopic[] getMqTopics() {
        MqTopic[] mqTopics = new MqTopic[1];
        mqTopics[0] = new MqTopic(OrderPayService.TOPIC_ORDER_STATUS_NOTIFY, Constants.CLUB_RECHARGE_CODE);
        return mqTopics;
    }

    /**
     * 查询俱乐部消费消息记录
     * @param msgId
     * @return
     */
    private List<TbClubMsgConsumerRecord> getClubMsgConsumerRecord(String msgId){
        TbClubMsgConsumerRecordExample example = new TbClubMsgConsumerRecordExample();
        example.createCriteria().andMsgIdEqualTo(msgId);
        List<TbClubMsgConsumerRecord> clubMsgConsumerRecordList = this.clubMsgConsumerRecordMapper.selectByExample(example);
        return clubMsgConsumerRecordList;
    }

    /**
     * 插入消息消费记录
     * @param record
     * @return
     */
    private int insertClubMsgComsumerRecord(TbClubMsgConsumerRecord record){
        int n = this.clubMsgConsumerRecordMapper.insertSelective(record);
        return n;
    }

    /**
     * 更新消息消费记录
     * @param record
     * @return
     */
    private int updateClubMsgComsumerRecord(TbClubMsgConsumerRecord record){
        int n = this.clubMsgConsumerRecordMapper.updateByPrimaryKeySelective(record);
        return n;
    }

}
